package com.zhao.streaming

import com.zhao.commonUtil.{ConnectionManager, JsonParser, MyUtils}
import com.zhao.util.{ParamParserUtil, SparkUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming.dstream.DStream

import java.sql
import java.sql.{Connection, PreparedStatement, Timestamp}
import java.util.{Calendar, Date, Timer, TimerTask}

/**
 * Description: 活跃用户统计<br/>
 * Copyright (c) ，2021 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2021/1/7 13:50
 *
 * @author 柒柒
 * @version : 1.0
 */

object ActiveUser {
  def main(args: Array[String]): Unit = {

    if (args.length < 1){
      System.err.println("参数不正确")
      System.exit(1)
    }

    printf("-----main-----args: " + args.mkString(" || "))
    Logger.getRootLogger.setLevel(Level.ERROR)

    //定时计算任务计算活跃用户数,30s刷新一次(2分钟后开始执行)
    calculateActive()
    //每天凌晨1点执行删除10天前的数据
    cleanData()

    //创建流
    val (messages,ssc) = SparkUtil.createKafkaStream(args)

    //messages.map(record =>(record.value().toString)).print

    //解析参数
    val paramMap: Map[String, String] = ParamParserUtil.argsToOptionMap(args)
    val toon_type = paramMap("toon_type")

    //过滤符合条件的字符串
    val prodMessages: DStream[ConsumerRecord[String, String]] = messages.filter(line => {
      val info: String = line.value()
      var flag = false
      //解析json
      try {
        val jsonParser: JsonParser = new JsonParser(info)
        if (info.contains("type") && info.contains("properties")
          && info.contains("toon_type") && info.contains("distinct_id")) {

          val distinct_id = jsonParser.getJsnField("distinct_id")
          if ("track".equals(jsonParser.getJsnField("type"))
            && toon_type.trim.equals(jsonParser.getSCJsnField("properties", "toon_type"))
            && distinct_id.trim.length <= 10
            && distinct_id.trim.length > 0) {
            flag = true
          }
        }
      } catch {
        case e: Throwable => e.printStackTrace()
      }
      flag
    })

    //解析userId
    val userDStream: DStream[String] = logParser(prodMessages)
    //打印
    //userDStream.print()
    persistMysql(userDStream)
    ssc.start()
    ssc.awaitTermination()
  }

  /**
   * 持久化到数据库
   * @param dstream
   */
  def persistMysql(dstream: DStream[String]) = {
    dstream.foreachRDD(rdd =>{
      rdd.foreachPartition(iter =>{
        val now = System.currentTimeMillis()
        var conn: Connection = null;
        var pstmt:PreparedStatement = null;
        try{
          conn = ConnectionManager.getConnnection
          conn.setAutoCommit(false)
          val sql =
            """
              |insert into fz_bigscreen_active_user_detail(user_id,update_time,stat_date)
              |values(?,?,?)
              |on duplicate key update user_id=VALUEs(user_id),update_time=VALUES(update_time),stat_date=VALUES(stat_date)
              |""".stripMargin

          pstmt = conn.prepareStatement(sql)
          while (iter.hasNext){
            val user_id = iter.next()
            pstmt.setString(1,user_id)
            pstmt.setTimestamp(2,new Timestamp(now))
            pstmt.setDate(3,new java.sql.Date(now))
            pstmt.addBatch()
          }
          pstmt.executeBatch()
          conn.commit()
        }catch{
          case e: Throwable => e.printStackTrace()
            conn.rollback()
        }finally {
          ConnectionManager.closeAll(null,pstmt,conn)
        }
      })
    })
  }

  def logParser(dstream: DStream[ConsumerRecord[String, String]]) = {
    dstream.map(line => {
      val info: String = line.value()
      val userId = MyUtils.getJsnField(info, "distinct_id")
      userId
    }).transform(rdd => rdd.distinct())
  }

  /**
   * 删除20天前的数据
   */
  def cleanData() = {
    val calendar = Calendar.getInstance()
    calendar.set(Calendar.HOUR_OF_DAY,1)
    calendar.set(Calendar.MINUTE,0)
    calendar.set(Calendar.SECOND,0)
    val time: Date = calendar.getTime()
    new Timer().scheduleAtFixedRate(new TimerTask(){
      override def run(): Unit = {

        val tenDayTime: Long = System.currentTimeMillis() - (86400 * 1000) * 10
        var conn: Connection = null
        var pstmt: PreparedStatement = null
        try{
          conn = ConnectionManager.getConnnection
          conn.setAutoCommit(false)
          val sql1 = "delete from fz_bigscreen_active_user_detail where stat_date <= ?"
          val sql2 = "delete from fz_bigscreen_user_info where stat_date <= ?"
          pstmt = conn.prepareStatement(sql1)
          pstmt.setDate(1,new sql.Date(tenDayTime))
          pstmt.execute()
          pstmt = conn.prepareStatement(sql2)
          pstmt.setDate(1,new sql.Date(tenDayTime))
          pstmt.execute()
          conn.commit()
        }catch {
          case e: Throwable => e.printStackTrace()
            conn.rollback()
        }finally {
          ConnectionManager.closeAll(null,pstmt,conn)
        }
      }
    },time,24 * 60 * 1000)
  }

  def calculateActive() = {
    val time: Date = new Date(System.currentTimeMillis() + (1000 * 60 * 2))

    new Timer().scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = {
        val now: Long = System.currentTimeMillis()
        val nowDate: sql.Date = new sql.Date(now)
        var conn: Connection = null
        var pstmt: PreparedStatement = null
        try{
          conn = ConnectionManager.getConnnection()
          conn.setAutoCommit(false)
          val sql =
            """
              |insert into fz_bigscreen_active_user_info(stat_date,stat_type,user_count,update_time)
              |values(?,'stream',select count(distinct user_id) from fz_bigscreen_active_user_detail as d where d.stat=?,?)
              |on duplicate key update user_count = (select count(distinct user_id) from fz_bigscreen_active_user_detail where stat_date=?),update_time=?
              |""".stripMargin

          pstmt = conn.prepareStatement(sql)
          pstmt.setDate(1,nowDate)
          pstmt.setDate(2,nowDate)
          pstmt.setTimestamp(3,new Timestamp(now))
          pstmt.setDate(4,nowDate)
          pstmt.setTimestamp(5,new Timestamp(now))
          pstmt.execute()
          conn.commit()
        }catch {
          case e: Throwable => e.printStackTrace()
            conn.rollback()
        }finally {
          ConnectionManager.closeAll(null,pstmt,conn)
        }
      }
    },time,30 * 1000)
  }
}













